package day08

import java.util.Properties

import kafka.producer.{KeyedMessage, Producer, ProducerConfig}

object ProducerDemo {
  def main(args: Array[String]): Unit = {
    /**
      * 实现一个生产者：
      *   模拟一些数据不断地发送到kafka的topic中
      *   实现自定义分区器
      *   根部，要创建Prodecer -> 需要参数ProducerConfig -> ProducerConfig需要参数Properties ->
      *     Properties.put方法来配置所需信息:eg : serializer.class=, metadata.broker.list=，request.required.acks=，partitioner.class=，其实不用设置，有默认的，根据自己需求来。
      *   producer.send(new KeyedMessage[String,String](topic, msg))生产消息
      */
    // 配置kafka的信息，创建properties类，该类用put方法专门制定k,v信息
    val props = new Properties()
    // 配置序列化类型,即producer.properties配置文件里serializer.class=kafka.serializer.DefaultEncoder，这里该为StringEncoder序列化
    props.put("serializer.class","kafka.serializer.StringEncoder")
    // 制定kafka的元数据存放节点，即producer.properties配置文件里的metadata.broker.list对应的节点和端口
    // 特别注意，像写这种字符串信息的时候，不要加多余的空格键。否则会解析不了出问题
    props.put("metadata.broker.list","master:9092")
    // 设置发送数据的响应方式。producer.properties配置文件里request.required.acks=
    props.put("request.required.acks","1")
    // 设置分区器为默认的分类器partitioner.class为kafka.producer.DefaultPartitione
    props.put("partitioner.class","kafka.producer.DefaultPartitioner")
    //props.put("partitioner.class","day08.MyPartitioner")
    // 指定topic名
    val topic = "test1"
    // 创建producer配置对象：ProducerConfig(配置信息)
    val config = new ProducerConfig(props)
    // 创建producer对象。将配置对象作为构造参数，注意把返回Producer[Nothing, Nothing]改为需要的Producer[String, String]
    // 注意：生成的数据 k 代表索引， v就是数据值。。。两者都是String
    val producer: Producer[String, String] = new Producer(config)
    //模拟数据，producer对象的send(参数是KeyedMessage可以new这个类把topic和数据传过来)方法发送数据
    for (i <- 1 to 10000) {
      val msg = s"$i: Producer send data"
      producer.send(new KeyedMessage[String,String](topic, msg))
      Thread.sleep(50)
    }
    // 如果一次性把数据发送到kafka，用producer.close()；
    // 如果是一直实时监控某个目录，就不能用producer.close()
    producer.close()
  }
}
